package com.lzwk.app.ods.ocpx;

import com.lzwk.beans.ocpx.DwdTest;
import com.lzwk.common.test.TestConstant;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @title: KafkaTest
 * @Author Yeaser
 * @Date: 2022/2/10 16:06
 * @Version 1.0
 * Description: TODO
 */
public class KafkaTest  {
    public static void main(String[] args) throws Exception {
/*        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 1,20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 1,50));
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 3,50));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 2,10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527991000L, 3,40));
        waterSensors.add(new WaterSensor("sensor_3", 1607527994000L, 1,35));
        waterSensors.add(new WaterSensor("sensor_3", 1607527995000L, 1,50));
        waterSensors.add(new WaterSensor("sensor_4", 1607527996000L, 2,40));
        waterSensors.add(new WaterSensor("sensor_4", 1607527999000L, 3,35));*/



        StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment()
                .setParallelism(1);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        tenv.executeSql("create table mysql_pay_order(" +
                " `id` INT, " +
                " `payer_id` INT, " +
                " `fee` INT, " +
                " `product_id` INT, " +
                " `source_id` INT, " +
                " `paid` INT, " +
                " `skin_id` INT, " +
                " `wxapp_id` INT, " +
                " `create_time` TIMESTAMP(0), " +
                " `pay_time` TIMESTAMP(0), " +
                " `meta` STRING NOT NULL, " +
                " `openid` STRING, " +
                " `proc_time` AS PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + TestConstant.TESTHOST + "', " +
                "   'port'='" + TestConstant.TESTPORT + "', " +
                "   'username'='" + TestConstant.TESTUSERNAME + "', " +
                "   'password'='" + TestConstant.TESTPASSWORD + "', " +
                "   'database-name'='" + TestConstant.TESTOCPXDB + "', " +
                "   'table-name'='" + TestConstant.TESTOCPXPAYORDER + "', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none' " +
                ")");

        Table table1 = tenv.sqlQuery(" select * from mysql_pay_order");

        tenv.createTemporaryView("test",table1);

        SingleOutputStreamOperator<DwdTest> sourceDS = tenv
                .toRetractStream(table1, DwdTest.class)
                .filter(data -> data.f0)
                .map(data -> data.f1)
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<DwdTest>forBoundedOutOfOrderness(Duration.ofDays(1))
                .withTimestampAssigner((data,time) -> data.getCreate_time().getTime()));




        Table table = tenv.fromDataStream(sourceDS, $("id"), $("payer_id"), $("fee"), $("product_id"), $("source_id"), $("paid"), $("skin_id"),
                $("wxapp_id"), $("create_time").rowtime(), $("pay_time"), $("meta"), $("openid"), $("proc_time"), $("pt").proctime());

        tenv.createTemporaryView("test1",table);
        /*SingleOutputStreamOperator<WaterSensor> sourceDS = env
                .fromCollection(waterSensors)
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                        .withTimestampAssigner((data, time) -> data.getTs()));*/



        //Table table = tenv.fromDataStream(sourceDS, $("id"), $("ts").rowtime(), $("tag"), $("vc"));

        //tenv.createTemporaryView("test",table);

        tenv.sqlQuery(" select *, sum(if(paid is not null,1,0)) over(partition by product_id,source_id,skin_id,if(DATE_FORMAT(create_time,'yyyy-MM-dd') < DATE_FORMAT(pay_time,'yyyy-MM-dd'),DATE_FORMAT(pay_time,'yyyy-MM-dd'),DATE_FORMAT(create_time,'yyyy-MM-dd')) order by create_time) as sum_order from test1").execute().print();


        env.execute();
    }
}
